package com.lagou.flink;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author: yehw
 * @date: 2021/1/21 下午10:42
 * @description: 自定义时间窗口函数
 */
public class MyTimeWindowFunction implements WindowFunction {
    @Override
    public void apply(Object o, Window window, Iterable iterable, Collector collector) throws Exception {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        TimeWindow timeWindow = (TimeWindow) window;
        long start = timeWindow.getStart();
        long end = timeWindow.getEnd();
        AtomicLong sum = new AtomicLong();
        iterable.forEach(x -> {
            Tuple2<String, Long> x1 = (Tuple2<String, Long>) x;
            sum.addAndGet(x1.f1);
        });
        Tuple2 tu = (Tuple2) o;
        collector.collect("key:" + tu.getField(0) + " value: " + sum +
                "| window_start :" + format.format(start) +
                " window_end :" + format.format(end));

    }
}
